package com.cw.realtime.dim.function

import com.cw.realtime.common.bean.TableProcessDim
import com.cw.realtime.common.constant.Constant.HBASE_NAMESPACE
import com.cw.realtime.common.util.HbaseUtil
import io.circe.parser.decode
import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.util.Collector
import org.apache.hadoop.hbase.client.Connection

class FlatMapFunction  extends RichFlatMapFunction[String, TableProcessDim]{

  var conn = Option.empty[Connection]

  override def open(parameters: Configuration): Unit =
    conn = HbaseUtil.getConnection

  override def close(): Unit = HbaseUtil.closeConnection(conn)

  override def flatMap(str: String, out: Collector[TableProcessDim]): Unit = {

    decode[TableProcessDim](str) match {
      case Left(err) => throw err
      case Right(tableProcessDim) =>
        tableProcessDim.op match {
          case "d" => dropTable(tableProcessDim)
          case "c" | "r" => createTable(tableProcessDim)
          case "u" =>
            dropTable(tableProcessDim); createTable(tableProcessDim)
        }
        out.collect(tableProcessDim)
    }
  }


  private def createTable(dim: TableProcessDim): Unit = {
    val families = dim.sinkFamily.split(",")
    conn.foreach { c => HbaseUtil.createTable(c, HBASE_NAMESPACE, dim.sinkTable, families: _*) }
  }


  private def dropTable(dim: TableProcessDim): Unit =
    conn.foreach { c => HbaseUtil.dropTable(c, HBASE_NAMESPACE, dim.sinkTable)
}


}
